package com.ziyucoding.springboot.labs.lab3.sca.rocketmqdemo.consumerdemo.listener;

import com.ziyucoding.springboot.labs.lab3.sca.rocketmqdemo.consumerdemo.message.Demo1Message;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Component;

/**
 * 示例1的消费者
 *
 * @author ziyucoding
 * @date 2022-10-24
 **/
@Component
public class Demo1Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @StreamListener(MySink.DEMO1_INPUT)
    public void onMessage(@Payload Demo1Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
        throw new RuntimeException("故意抛出异常，模拟消费失败");
    }

    /**
     * 局部异常处理
     *
     * @param errorMessage
     * @return void
     **/
    @ServiceActivator(inputChannel = "demo1-topic.demo1-consumer-group-demo1-topic.errors")
    public void handleError(ErrorMessage errorMessage) {
        logger.error("[handleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        logger.error("[handleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        logger.error("[handleError][headers：{}]", errorMessage.getHeaders());
    }

    /**
     * 全局异常处理
     *
     * @param errorMessage
     * @return void
     **/
    @StreamListener(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
    public void globalHandleError(ErrorMessage errorMessage) {
        logger.error("[globalHandleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        logger.error("[globalHandleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        logger.error("[globalHandleError][headers：{}]", errorMessage.getHeaders());
    }
}
